package com.lingjtx.iot.network.config;

import com.lingjtx.common.core.constant.MqConst;
import com.lingjtx.common.redis.config.ServerNodeConfig;
import com.lingjtx.iot.network.common.MqttPropertiesEntity;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

/**
 * 公共配置
 */
@Slf4j
@Order(1)
@Configuration
public class MqttConfig {

    private final ServerNodeConfig serverNodeConfig;

    public MqttConfig(ServerNodeConfig serverNodeConfig) {
        this.serverNodeConfig = serverNodeConfig;
    }

    @Bean
    @ConfigurationProperties(prefix = "mqtt")
    public MqttPropertiesEntity mqttProperties() {
        return new MqttPropertiesEntity();
    }

    @Bean("mqttClientId")
    public String mqttClientId() {
        return "mqtt_" + serverNodeConfig.getNodeId();
    }

    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{mqttProperties().getBrokerUrl()});
        options.setUserName(mqttProperties().getUsername());
        options.setPassword(mqttProperties().getPassword().toCharArray());
        options.setConnectionTimeout(mqttProperties().getTimeout());
        options.setKeepAliveInterval(mqttProperties().getKeepAlive());
        options.setCleanSession(false);
        options.setAutomaticReconnect(true);
        options.setMaxInflight(mqttProperties().getMaxInflight()); // 最大重试次数
        options.setMaxReconnectDelay(mqttProperties().getMaxReconnectDelay()); // 最大重连间隔,单位：毫秒
        options.setWill(String.format(MqConst.MqttPublisherTopic.ACTION_SYS_SERVER, mqttClientId()),
                "offline".getBytes(),
                mqttProperties().getQos(),
                false);
        return options;
    }

    @Bean
    public TaskScheduler taskScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(Runtime.getRuntime().availableProcessors()); // 根据实际负载调整线程池大小
        scheduler.setThreadNamePrefix("mqtt-task-");
        scheduler.setWaitForTasksToCompleteOnShutdown(true);
        scheduler.initialize();
        return scheduler;
    }

    @Bean
    public RetryTemplate retryTemplate() {
        return RetryTemplate.builder()
                .maxAttempts(mqttProperties().getRetry().getMaxAttempts())
                .exponentialBackoff(mqttProperties().getRetry().getInitialInterval(),
                        mqttProperties().getRetry().getMultiplier(),
                        mqttProperties().getRetry().getMaxInterval())
                .retryOn(MqttException.class)
                .traversingCauses()
                .withListener(new RetryListener() {
                    @Override
                    public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
                        RetryListener.super.onError(context, callback, throwable);
                        log.warn("retryTemplate-> MQTT发送重试中，次数：{}", context.getRetryCount());
                    }
                })
                .build();
    }
}
